package com.burges.net.ml.dataStructure

import org.apache.flink.api.java.operators.DataSink
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
import org.apache.flink.ml.MLUtils
import org.apache.flink.ml.common.LabeledVector
import org.apache.flink.ml.math.DenseVector

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/26 
  * 描述     FlinkML构建Vector数据两种方式
  */
object MakeVectorData {

	def main(args: Array[String]): Unit = {
		val environment = ExecutionEnvironment.getExecutionEnvironment
		/**
		  * 通过LibSVM数据
		  */
		//读取LibSVM数据
		MLUtils.readLibSVM(environment, "/path/svmfile")
		MLUtils.readLibSVM(environment, "/path/svmfile2")
		//写出LibSVM数据
		val svmData: DataSet[LabeledVector] = environment.fromElements(LabeledVector(1.0, DenseVector(1)))
		val dataSink: DataSink[String] = MLUtils.writeLibSVM("/path/svmfile2", svmData)

		/**
		  * 读取CSV文件转换
		  */
		val trainCsvData = environment.readCsvFile[(String, String, String, String)]("/path/svm_train.data")
		val trainData:DataSet[LabeledVector] = trainCsvData.map {
			data =>
				val numList = data.productIterator.toList.map(_.asInstanceOf[String].toDouble)
			LabeledVector(numList(3), DenseVector(numList.take(3).toArray))
		}
	}

}
